package com.smsc.headend.connector.service.impl;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.convert.Convert;
import cn.hutool.core.date.SystemClock;
import cn.hutool.core.lang.UUID;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.google.common.cache.*;
import com.smsc.headend.common.utils.RedisKeys;
import com.smsc.headend.common.utils.RedisUtils;
import com.smsc.headend.connector.enums.consts.NettyConstant;
import com.smsc.headend.connector.bo.TaskContext;
import com.smsc.headend.connector.manager.DeviceKeyManager;
import com.smsc.headend.connector.netty.NettyConfigHelper;
import com.smsc.headend.connector.netty.SyncFuture;
import com.smsc.headend.connector.service.*;
import com.smsc.headend.connector.utils.NetUtil;

import com.smsc.headend.module.asset.entity.AnonyDevice;
import com.smsc.headend.module.asset.entity.ComEntity;
import com.smsc.headend.module.data.dto.ClearLocalCacheDto;
import com.smsc.headend.module.task.consts.TaskKafkaTopic;
import io.netty.channel.Channel;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.time.DateUtils;
import org.assertj.core.util.Lists;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import java.nio.charset.Charset;
import java.time.Instant;
import java.time.ZoneId;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantReadWriteLock;

@Service
@Slf4j
public class TerminalChannelServiceImpl implements TerminalChannelService {
    public static final String LOGIN = "login:";
    @Autowired
    LogTerminalAccessService logTerminalAccessService;

    @Autowired
    RedisUtils redisUtils;
    @Autowired
    AnonyDeviceService anonyDeviceService;
    @Autowired
    FeignAssetManagementService assetManagementService;
    @Autowired
    DeviceStatusService deviceStatusService;
    @Autowired
    DeviceKeyManager deviceKeyManager;

    private Map<String, TaskContext> lastSendContext = new ConcurrentHashMap<>();

    @Autowired
    KafkaTemplate kafkaTemplate;

    ReentrantReadWriteLock channelTerminalLock = new ReentrantReadWriteLock();

    //缓存返回报文，用于同步相应。 如AA建立
    private static Cache<String, SyncFuture> syncFutureMap = CacheBuilder.newBuilder()
            //设置缓存容器的初始容量为10
            .initialCapacity(100)
            // maximumSize 设置缓存大小
            .maximumSize(65535)
            //设置并发级别为20，并发级别是指可以同时写缓存的线程数
            .concurrencyLevel(200)
            // expireAfterWrite设置写缓存后121秒钟过期
            .expireAfterWrite(121, TimeUnit.SECONDS)
            //设置缓存的移除通知
            .removalListener(new RemovalListener<Object, Object>() {
                @Override
                public void onRemoval(RemovalNotification<Object, Object> notification) {
                    log.debug("AA LoadingCache: {} was removed, cause is {}", notification.getKey(), notification.getCause());
                }
            })
            .build();

    //缓存接口这里是LoadingCache，LoadingCache在缓存项不存在时可以自动加载缓存
    private static Cache<String, Map<String, TaskContext>> channelTaskContextCache = CacheBuilder.newBuilder()
            //设置缓存容器的初始容量
            .initialCapacity(100)
            // maximumSize 设置缓存大小
            .maximumSize(65535)
            //设置并发级别为20，并发级别是指可以同时写缓存的线程数
            .concurrencyLevel(100)
            // expireAfterWrite设置写缓存后60秒过期
            .expireAfterWrite(60 * 60, TimeUnit.SECONDS)
            //设置缓存的移除通知
            .removalListener(new RemovalListener<Object, Object>() {
                @Override
                public void onRemoval(RemovalNotification<Object, Object> notification) {
                    log.debug("Channel LoadingCache: {} was removed, cause is {}", notification.getKey(), notification.getCause());
                }
            }).build();

    LoadingCache<Long, ComEntity> comEntityLocalCache = CacheBuilder.newBuilder()
            //设置缓存容器的初始容量为10
            .initialCapacity(1000)
            // maximumSize 设置缓存大小
            .maximumSize(100000)
            //设置并发级别为20，并发级别是指可以同时写缓存的线程数
            .concurrencyLevel(20)
            .expireAfterAccess(99, TimeUnit.MINUTES)
            .build(new CacheLoader<Long, ComEntity>() {
                @Override
                public ComEntity load(Long meterId) throws Exception {
                    return assetManagementService.getComById(meterId);
                }
            });

    ConcurrentHashMap<Long, Long> comOnlineTime = new ConcurrentHashMap();

    @Override
    public boolean isTerminalInCurrentNode(Long comId) {
        Channel channel = NettyConfigHelper.getChannelByComId(comId);
        if (channel == null || !channel.isActive()) {
            return false;
        }
        return true;
    }

    @Override
    public void addReceiveRecord(ComEntity com) {
        comOnlineTime.put(com.getComId(), SystemClock.now() / DateUtils.MILLIS_PER_SECOND);
    }

    @Override
    public void disconnectDeprecatedChannel(String moduleSN, String loginCode) {
        List<Channel> allChannelByComId = NettyConfigHelper.getAllChannelByModuleSN(moduleSN);
        log.debug("getChannelInThisNode:sn:{},size:{}", moduleSN, Optional.of(allChannelByComId).orElse(Lists.emptyList()).size());
        if (CollectionUtil.isEmpty(allChannelByComId)) {
            return;
        }
        log.debug("currentTCPConnection:num={},moduleSN={}", allChannelByComId.size(), moduleSN);
        allChannelByComId.stream().filter(o -> o.attr(NettyConfigHelper.loginCode).get() != null && !o.attr(NettyConfigHelper.loginCode).get().equalsIgnoreCase(loginCode))
                .forEach(
                        channel -> {
                            ComEntity com = channel.attr(NettyConfigHelper.comEntityAttrKey).get();
                            log.info("disconnectTcpChannel:sn={},comId={},netaddr={}", moduleSN, com.getComId(), NetUtil.getIpFromChannel(channel));
                            channel.attr(NettyConfigHelper.closeBySystem).set(Boolean.TRUE);
                            NettyConfigHelper.removeChannel(channel);
                            try {
                                channel.close();
                            } catch (Exception e) {
                                log.error("chnClsFailed0", e);
                            }
                        }
                );
    }

    @Override
    public void checkOfflineComEntity() {
        log.info("start check offline com unit");
        Long startTime = SystemClock.now();
        Long now = SystemClock.now() / 1000;
        for (Long comId : comOnlineTime.keySet()) {
            Long lastReceiveTime = comOnlineTime.get(comId);
            if (lastReceiveTime != null && lastReceiveTime > 0 && now - lastReceiveTime > 10 * 60 && heartbeatInRedisOverTime(comId)) {
                log.warn("comId:{},lastReceive:{}", comId, Instant.ofEpochSecond(lastReceiveTime).atZone(ZoneId.systemDefault()));
                lostConnect(comId);
            }
        }
        log.info("end check offline com unit, cost: {}ms", SystemClock.now() - startTime);
    }

    private boolean heartbeatInRedisOverTime(Long comId) {
        String heartbeatTimeStr = Convert.toStr(redisUtils.get(RedisKeys.getComLastHeartbeat(comId)));
        if (heartbeatTimeStr == null) {
            return true;
        }
        Long heartbeatTime = Convert.toLong(heartbeatTimeStr);
        if (heartbeatTime + 10 * 60 < SystemClock.now() / DateUtils.MILLIS_PER_SECOND) {
            return true;
        }
        return false;
    }

    private void lostConnect(Object comId) {
        Long comEntityId = Convert.toLong(comId);
        comOnlineTime.remove(comId);
        Channel channel = NettyConfigHelper.getChannelByComId(comEntityId);
        if (channel == null) {
            return;
        }
        Long loginTime = channel.attr(NettyConfigHelper.deviceLgcTime).get();
        if (SystemClock.now() - loginTime > DateUtils.MILLIS_PER_HOUR) {
            log.info("channelNoMessage systemClosing..comId：{}", comEntityId);
            NettyConfigHelper.disconnectTcp(NetUtil.getIpFromChannel(channel));
        }
    }

    @Override
    public String getNetAddressByTerminal(ComEntity terminal) {
        Channel channelByComId = NettyConfigHelper.getChannelByComId(terminal.getComId());
        return NetUtil.getIpFromChannel(channelByComId);
    }

    @Override
    public void addTerminalNetAddress(ComEntity comEntity, Channel channel) {
        log.debug("addTerminalNetAddress comEntity={}, netAddr={}", comEntity, channel);
        String loggingCode = UUID.fastUUID().toString();
        channel.attr(NettyConfigHelper.loginCode).set(loggingCode);
        String closeDeprecatedConnection = LOGIN + comEntity.getModuleSn() + ":" + loggingCode;
        kafkaTemplate.send(TaskKafkaTopic.CLOSE_ANONY_ASSET_CHANNEL, closeDeprecatedConnection.getBytes(Charset.forName("utf-8")));
        if (comEntity.getComId() != NettyConstant.ANONYMOUS_DEFAULT_COM_ID && comEntity.getComMode() == ComEntity.CommunicationMode.SERVER.getCode()) {
            channel.attr(NettyConfigHelper.signMsgSent).set(true);
            assetManagementService.updateComEntityStatus(comEntity.getComId(), 1, SystemClock.now() / 1000);
            log.info("COM registered: id->{}", comEntity.getComId());
            deviceStatusService.refreshDeviceStatus(comEntity, null, ComEntity.CommunicationStatus.ONLINE);
        }
        //kickOff Same ModuleSn Device
    }

    @Override
    public byte[] getSyncResponse(Channel channel, Integer getTimeout, String deviceAddress) throws InterruptedException {
        if (StringUtils.isEmpty(deviceAddress)) {
            deviceAddress = "-1";
        }
        String channelIpAddress = NetUtil.getIpFromChannel(channel);
        SyncFuture responseFuture = new SyncFuture();
        syncFutureMap.put(channelIpAddress + "-" + deviceAddress, responseFuture);
        byte[] bytes = responseFuture.get(getTimeout, TimeUnit.SECONDS);
        syncFutureMap.invalidate(channelIpAddress + "-" + deviceAddress);
        return bytes;
    }


    @Override
    public SyncFuture getFutureSync(String netAddress, String deviceAddress) {
        if (StringUtils.isEmpty(deviceAddress)) {
            deviceAddress = "-1";
        }
        return syncFutureMap.getIfPresent(netAddress + "-" + deviceAddress);
    }


    @Override
    public ComEntity getTerminalByNetAddress(String netAddress) {
        Channel channelByAddress = NettyConfigHelper.getChannelByAddress(netAddress);
        if (channelByAddress == null) {
            return null;
        }
        return channelByAddress.attr(NettyConfigHelper.comEntityAttrKey).get();
    }

    @Override
    public TaskContext getTaskContext(String netAddress, String meterNo, Long termId) {
        Map<String, TaskContext> channelContexts = channelTaskContextCache.getIfPresent(netAddress);
        if (channelContexts == null) {
            return null;
        }
        String key = null;
        if (meterNo == null) {
            key = String.valueOf(termId);
        } else {
            key = String.format("%s:%s", termId, meterNo);
        }
        return channelContexts.get(key);
    }

    public void putNetAddressTaskContext(String netAddress, TaskContext taskContext) {
        Map<String, TaskContext> channelContexts = channelTaskContextCache.getIfPresent(netAddress);
        if (channelContexts == null) {
            channelContexts = new HashMap();
        }
        String key = null;
        if (taskContext.getDeviceAddress() == null) {
            key = String.valueOf(taskContext.getComId());
        } else {
            key = String.format("%s:%s", taskContext.getComId(), taskContext.getDeviceAddress());
        }
        channelContexts.put(key, taskContext);
        channelTaskContextCache.put(netAddress, channelContexts);
    }

    @Override
    public void closeAnonyChannel(String moduleSn) {
        anonyDeviceService.remove(Wrappers.<AnonyDevice>lambdaQuery().eq(AnonyDevice::getDeviceSn, moduleSn));
        ComEntity comEntity = new ComEntity();
        comEntity.setModuleSn(moduleSn);
        comEntity.setComId(NettyConstant.ANONYMOUS_DEFAULT_COM_ID);
        Channel channel = NettyConfigHelper.getChannelByIdAndModuleSN(comEntity);
        if (channel != null) {
            log.info("foundChannelKickingOffAnonyDeviceInSn:{}", moduleSn);
            NettyConfigHelper.removeChannel(channel);
            channel.close();
        }
    }

    @Override
    public void setChannelLastSendContext(String netAddress, TaskContext context) {
        log.debug("context last: set netaddr={}, context={}", netAddress, context);
        lastSendContext.put(netAddress, context);
    }

    @Override
    public TaskContext getChannelLastSendContext(String netAddress) {
        return lastSendContext.get(netAddress);
    }

    @Override
    public void removeChannelLastContext(String netAddress) {
        lastSendContext.remove(netAddress);
    }

    @Override
    public ComEntity getTerminalById(Long terminalId) {
        if (-1L == terminalId) {
            return null;
        }
        try {
            return comEntityLocalCache.get(terminalId);
        } catch (ExecutionException e) {
            log.error("get cached comEntity {} fail");
        }
        return null;
    }

    @Override
    public void clearComEntityCacheById(ClearLocalCacheDto dto) {
        if (dto.getComStatus() == null) {

            ComEntity cachedCom = comEntityLocalCache.getIfPresent(dto.getComId());
            if (cachedCom != null) {
                deviceKeyManager.invalidKey(cachedCom.getDeviceId());
            }
            comEntityLocalCache.invalidate(dto.getComId());
        } else {
            ComEntity cachedCom = comEntityLocalCache.getIfPresent(dto.getComId());
            if (cachedCom != null) {
                cachedCom.setComStatus(dto.getComStatus());
                comEntityLocalCache.put(dto.getComId(), cachedCom);
            }
        }
    }

}
